package service

import (
	"errors"
	"gitee.com/sunny-zhangqiang/zcache_new/cache/base"
	"gitee.com/sunny-zhangqiang/zcache_new/comm"
	"gitee.com/sunny-zhangqiang/zcache_new/config"
	"gitee.com/sunny-zhangqiang/zcache_new/logger"
	"gitee.com/sunny-zhangqiang/zcache_new/ztp"
	"strconv"
	"time"
)

const (
	sourceKeyHeaderName             = "ZCache-SourceKey"
	sourceVersionHeaderName         = "ZCache-SourceVersion"
	sourceIdentifyHeaderName        = "ZCache-SourceIdentify"
	clientTokenHeaderName           = "ZCache-ClientToken"
	ifNoneMatchHeaderName           = "ZCache-IfNoneMatch"
	realServerHeaderName            = "ZCache-RealServer"
	cacheStatusHeaderName           = "ZCache-Status"
	errorHeaderName                 = "ZCache-Error"
	mergeTimeoutMSecHeaderName      = "ZCache-MergeTimeoutMSec"
	clientNotifyHeaderName          = "ZCache-ClientNotify"
	mergeTokenHeaderName            = "ZCache-MergeToken"
	mergeErrorHeaderName            = "ZCache-MergeError"
	cachableHeaderName              = "ZCache-Cacheable"
	startChunkMergeHeaderName       = "ZCache-StartChunkMerge"
	chunkMergeTimeoutMSecHeaderName = "ZCache-ChunkMergeTimeoutMSec"
	chunkFragmentSizeHeaderName     = "ZCache-ChunkFragmentSize"
	chunkMergeTokenHeaderName       = "ZCache-ChunkMergeToken"
	revalidStatusHeaderName         = "ZCache-Revalidate-Status"
	headerSourceKeyHeaderName       = "ZCache-HeaderSourceKey"
	chunkMergeErrorHeaderName       = "ZCache-ChunkMergeError"
	chunkFragmentIndexHeaderName    = "ZCache-ChunkFragmentIndex"
	chunkMergeDoneHeaderName        = "ZCache-ChunkMergeDone"
	chunkMergeJoinHeaderName        = "ZCache-ChunkMergeJoin"
	chunkMergeReleaseHeaderName     = "ZCache-ChunkMergeRelease"
	sourceContentLengthHeaderName   = "ZCache-SourceContentLength"
)

var (
	noBody = make([]byte, 0)
)

type request struct {
	c    *ztp.Conn
	req  *ztp.Request
	resp *ztp.Response

	sourceKey      string
	sourceVersion  string
	sourceIdentify string
	clientToken    string
	ifNoneMatch    string
	realServer     string

	mergeToken string
	mergeError string

	headerSourceKey    string
	chunkMergeError    string
	chunkFragmentSize  string
	chunkMergeDone     string
	chunkFragmentIndex string
	chunkMergeToken    string
	chunkMergeJoin     string
	chunkMergeRelease  string

	cachable            bool
	startChunkMergeFlag bool

	err error
}

func requestInit(r *request, c *ztp.Conn, req *ztp.Request, resp *ztp.Response) {
	r.c = c
	r.req = req
	r.resp = resp
	r.sourceKey = r.req.Headers[sourceKeyHeaderName]
	r.clientToken = r.req.Headers[clientTokenHeaderName]
}

func (r *request) done() {
	if err := recover(); err != nil {
		logger.Error("PanicErr: %+v in %s", err, r.req.LogSuffix)
		r.c.Closed = true
	}

	printAccessLog(r)
}

func (r *request) getInRealServer() (value []byte, err error) {
	if r.realServer == "" {
		return nil, noRealServerErr
	}

	req := ztp.Request{}
	req.Headers = make(map[string]string)
	resp := ztp.Response{}
	resp.Headers = make(map[string]string)

	req.Host = r.req.Host
	req.Action = "/get"
	req.TraceID = r.req.TraceID

	err = srv.client.Do(r.realServer, &req, &resp)
	if err != nil {
		return nil, err
	}

	if resp.Status != 200 ||
		resp.Headers[cacheStatusHeaderName] != "HIT" {
		return nil, cacheMissErr
	}

	if resp.Body == nil {
		resp.Body = noBody
	}

	return resp.Body, nil
}

func (r *request) get() {
	defer r.done()

	if r.sourceKey == "" {
		r.resp.Headers[errorHeaderName] = noSourceKeyHeaderErr.Error()
		r.resp.Status = 400
		return
	}

	r.realServer = r.req.Headers[realServerHeaderName]
	r.ifNoneMatch = r.req.Headers[ifNoneMatchHeaderName]
	key := base.GenObjectKey(comm.String2Bytes(r.sourceKey))
	value, err := cacher.Get(key)
	if err != nil && r.realServer != "" {
		value, err = r.getInRealServer()
		if err == nil {
			cacher.Put(key, value)
		}
	}

	if err != nil {
		r.resp.Status = 200
		return
	}

	if r.ifNoneMatch == "" {
		r.resp.Headers[cacheStatusHeaderName] = "HIT"
		r.resp.Status = 200
		r.resp.Body = value
		return
	}

	obj := cacheObject{}
	err = ParseCacheObject(&obj, value)
	if err != nil {
		cacher.Del(key)
		r.resp.Status = 200
		return
	}

	r.resp.Headers[cacheStatusHeaderName] = "HIT"

	r.sourceVersion = r.resp.Headers[sourceVersionHeaderName]
	if r.sourceVersion != "" && r.sourceVersion == obj.version {
		r.resp.Status = 304
		return
	}

	r.sourceIdentify = r.resp.Headers[sourceIdentifyHeaderName]
	if r.sourceIdentify != "" && r.sourceIdentify == obj.identify {
		r.resp.Status = 304
		return
	}

	r.resp.Status = 200
	r.resp.Body = value
	return
}

func (r *request) put() {
	defer r.done()

	if r.sourceKey == "" {
		r.resp.Headers[errorHeaderName] = noSourceKeyHeaderErr.Error()
		r.resp.Status = 400
		return
	}

	if r.req.Headers[cachableHeaderName] == "" {
		r.resp.Headers[errorHeaderName] = "no ZCache-Cacheable header"
		r.resp.Status = 400
		return
	}

	key := base.GenObjectKey(comm.String2Bytes(r.sourceKey))

	if r.req.Body == nil {
		r.req.Body = noBody
	}

	cacher.Put(key, r.req.Body)

	r.resp.Status = 200
	return
}

func (r *request) del() {
	defer r.done()

	if r.sourceKey == "" {
		r.resp.Headers[errorHeaderName] = noSourceKeyHeaderErr.Error()
		r.resp.Status = 400
		return
	}

	key := base.GenObjectKey(comm.String2Bytes(r.sourceKey))
	cacher.Del(key)

	r.resp.Status = 200
	return
}

func (r *request) getAndWait() {
	defer r.done()

	if r.sourceKey == "" {
		r.resp.Headers[errorHeaderName] = noSourceKeyHeaderErr.Error()
		r.resp.Status = 400
		return
	}

	if r.clientToken == "" {
		r.resp.Headers[errorHeaderName] = noClientTokenHeaderErr.Error()
		r.resp.Status = 400
		return
	}

	mergeTimeoutMSec := config.ServerConf.MergeTimeoutMSec
	mergeTimeoutMSecHeader := r.req.Headers[mergeTimeoutMSecHeaderName]
	if mergeTimeoutMSecHeader != "" {
		i, err := strconv.ParseInt(mergeTimeoutMSecHeader, 10, 64)
		if err == nil && i > 0 {
			mergeTimeoutMSec = i
		}
	}

	r.ifNoneMatch = r.req.Headers[ifNoneMatchHeaderName]
	r.sourceVersion = r.resp.Headers[sourceVersionHeaderName]
	r.sourceIdentify = r.resp.Headers[sourceIdentifyHeaderName]

	key := base.GenObjectKey(comm.String2Bytes(r.sourceKey))
	c, token := mge.wait(key, r.clientToken, mergeTimeoutMSec)

	if c != nil {
		md := <-c
		if md.clientNotify {
			r.resp.Headers[clientNotifyHeaderName] = "1"
		}

		if md.err != nil {
			if md.err != base.ErrNotFound {
				r.resp.Headers[errorHeaderName] = md.err.Error()
			}
			r.resp.Status = 200
			return
		}

		if !md.cachable {
			r.resp.Status = 200
			return
		}

		if r.ifNoneMatch == "" {
			r.resp.Headers[cacheStatusHeaderName] = "HIT"
			r.resp.Status = 200
			r.resp.Body = md.value
		}

		obj := cacheObject{}
		err := ParseCacheObject(&obj, md.value)
		if err != nil {
			cacher.Del(key)
			r.resp.Status = 200
			return
		}

		r.resp.Headers[cacheStatusHeaderName] = "HIT"

		if r.sourceVersion != "" && r.sourceVersion == obj.version {
			r.resp.Status = 304
			return
		}
		if r.sourceIdentify != "" && r.sourceIdentify == obj.identify {
			r.resp.Status = 304
			return
		}

		r.resp.Status = 200
		r.resp.Body = md.value
		return
	}

	r.realServer = r.req.Headers[realServerHeaderName]
	value, err := cacher.Get(key)
	if err != nil && r.realServer != "" {
		value, err = r.getInRealServer()
		if err == nil {
			cacher.Put(key, value)
		}
	}

	if err != nil {
		r.resp.Headers[mergeTokenHeaderName] = token
		r.resp.Status = 200
		return
	}

	mge.notify(key,
		mergeData{false, true, value, "", err},
		token)

	if r.ifNoneMatch == "" {
		r.resp.Headers[cacheStatusHeaderName] = "HIT"
		r.resp.Status = 200
		r.resp.Body = value
		return
	}

	obj := cacheObject{}
	err = ParseCacheObject(&obj, value)
	if err != nil {
		cacher.Del(key)
		r.resp.Status = 200
		return
	}

	r.resp.Headers[cacheStatusHeaderName] = "HIT"

	if r.sourceVersion != "" && r.sourceVersion == obj.version {
		r.resp.Status = 304
		return
	}
	if r.sourceIdentify != "" && r.sourceIdentify == obj.identify {
		r.resp.Status = 304
		return
	}

	r.resp.Status = 200
	r.resp.Body = value
	return
}

func (r *request) startChunkMerge() error {
	r.startChunkMergeFlag = true

	if r.sourceVersion == "" {
		return noSourceVersionHeaderErr
	}

	if r.clientToken == "" {
		return noClientTokenHeaderErr
	}

	chunkMergeTimeoutMSec := config.ServerConf.ChunkMergeTimeoutMSec
	chunkMergeTimeoutMSecHeader := r.req.Headers[chunkMergeTimeoutMSecHeaderName]
	if chunkMergeTimeoutMSecHeader != "" {
		i, err := strconv.ParseInt(chunkMergeTimeoutMSecHeader, 10, 64)
		if err == nil && i > 0 {
			chunkMergeTimeoutMSec = i
		}
	}

	chunkFragmentSize := int64(0)
	r.chunkFragmentSize = r.req.Headers[chunkFragmentSizeHeaderName]
	if r.chunkFragmentSize != "" {
		i, err := strconv.ParseInt(r.chunkFragmentSize, 10, 64)
		if err == nil && i > 0 {
			chunkFragmentSize = i
		}
	}

	if chunkFragmentSize == 0 {
		return invalidChunkFragmentSizeHeaderErr
	}

	cmKey := r.sourceKey + "#" + r.sourceVersion
	chunkMergeToken := cm.insert(cmKey, r.clientToken,
		chunkMergeData{
			fragmentSize:  chunkFragmentSize,
			fragmentIndex: 0,
		},
		chunkMergeTimeoutMSec)

	r.resp.Headers[chunkMergeTokenHeaderName] = chunkMergeToken
	return nil
}

func (r *request) putAndNotify() {
	defer r.done()

	if r.sourceKey == "" {
		r.resp.Headers[errorHeaderName] = noSourceKeyHeaderErr.Error()
		r.resp.Status = 400
		return
	}

	r.sourceVersion = r.req.Headers[sourceVersionHeaderName]

	r.mergeToken = r.req.Headers[mergeTokenHeaderName]
	if r.mergeToken == "" {
		r.resp.Headers[mergeTokenHeaderName] = noMergeTokenErr.Error()
		r.resp.Status = 400
		return
	}

	r.mergeError = r.req.Headers[mergeErrorHeaderName]
	var mergeErrorErr error
	if r.mergeError == "" {
		mergeErrorErr = errors.New(r.mergeError)
	}

	if r.req.Headers[startChunkMergeHeaderName] != "" {
		err := r.startChunkMerge()
		if err != nil {
			r.resp.Headers[errorHeaderName] = err.Error()
			r.resp.Status = 400
			return
		}
	}

	key := base.GenObjectKey(comm.String2Bytes(r.sourceKey))
	if r.req.Headers[cachableHeaderName] != "" {
		r.cachable = true
	}

	if r.req.Body == nil {
		r.req.Body = noBody
	}

	if r.cachable {
		cacher.Put(key, r.req.Body)
	}

	mge.notify(key,
		mergeData{true, r.cachable, r.req.Body, "", mergeErrorErr},
		r.mergeToken)

	r.resp.Status = 200
}

func (r *request) revalidateWait() {
	defer r.done()

	if r.sourceKey == "" {
		r.resp.Headers[errorHeaderName] = noSourceKeyHeaderErr.Error()
		r.resp.Status = 400
		return
	}

	if r.clientToken == "" {
		r.resp.Headers[errorHeaderName] = noClientTokenHeaderErr.Error()
		r.resp.Status = 400
		return
	}

	r.sourceVersion = r.req.Headers[sourceVersionHeaderName]
	r.sourceIdentify = r.req.Headers[sourceIdentifyHeaderName]

	if r.sourceVersion == "" && r.sourceIdentify == "" {
		r.resp.Headers[errorHeaderName] = "no sourceVersion and sourceIdentify"
		r.resp.Status = 400
		return
	}

	mergeTimeoutMSec := config.ServerConf.MergeTimeoutMSec
	mergeTimeoutMSecHeader := r.req.Headers[mergeTimeoutMSecHeaderName]
	if mergeTimeoutMSecHeader == "" {
		i, err := strconv.ParseInt(mergeTimeoutMSecHeader, 10, 64)
		if err == nil && i > 0 {
			mergeTimeoutMSec = i
		}
	}

	key := base.GenObjectKey(comm.String2Bytes(
		r.sourceKey + "#revalidate#" + r.sourceVersion + r.sourceIdentify))
	c, token := mge.wait(key, r.clientToken, mergeTimeoutMSec)
	if c != nil {
		md := <-c
		if md.clientNotify {
			r.resp.Headers[clientNotifyHeaderName] = "1"
		}

		if md.err != nil {
			r.resp.Headers[errorHeaderName] = md.err.Error()
			r.resp.Status = 200
			return
		}

		if md.revalidateStatus != "" {
			r.resp.Headers[revalidStatusHeaderName] = md.revalidateStatus
		}

		r.resp.Status = 200
		return
	}

	r.resp.Headers[mergeTokenHeaderName] = token
	r.resp.Status = 200
}

func (r *request) revalidateNotify() {
	defer r.done()

	if r.sourceKey == "" {
		r.resp.Headers[errorHeaderName] = noSourceKeyHeaderErr.Error()
		r.resp.Status = 400
		return
	}

	if r.clientToken == "" {
		r.resp.Headers[errorHeaderName] = noClientTokenHeaderErr.Error()
		r.resp.Status = 400
		return
	}

	r.sourceVersion = r.req.Headers[sourceVersionHeaderName]
	r.sourceIdentify = r.req.Headers[sourceIdentifyHeaderName]
	if r.sourceVersion == "" && r.sourceIdentify == "" {
		r.resp.Headers[errorHeaderName] = "no sourceVersion and sourceIdentify"
		r.resp.Status = 400
		return
	}

	r.mergeToken = r.req.Headers[mergeTokenHeaderName]
	if r.mergeToken == "" {
		r.resp.Headers[errorHeaderName] = noMergeTokenErr.Error()
		r.resp.Status = 400
		return
	}

	mergeErr := r.req.Headers[mergeErrorHeaderName]
	var mergeErrorErr error
	if mergeErr == "" {
		mergeErrorErr = errors.New(mergeErr)
	}

	revalidStatus := r.req.Headers[revalidStatusHeaderName]
	if revalidStatus == "" {
		r.resp.Headers[errorHeaderName] = noRevalidateStatusHeaderErr.Error()
		r.resp.Status = 400
		return
	}

	key := base.GenObjectKey(comm.String2Bytes(
		r.sourceKey + "#revalidate#" + r.sourceVersion + r.sourceIdentify))

	// ************************* ******************************
	// TODO: 如果想要保证缓存的强一致性, 建议FusionCache提供一个事物接口.
	//	cacher.(*FusionCache).Transaction(func() {
	//		cacher.unsafeGet(...)
	//		parseCacheObject & update & SerialCacheObject
	//		cacher.unsafePut(...)
	//	})

	if revalidStatus == "304" {
		canUpdate := true
		obj := cacheObject{}
		value, err := cacher.Get(key)
		if err != nil {
			canUpdate = false
		} else {
			err = ParseCacheObject(&obj, value)
			if err != nil {
				canUpdate = false
			}
		}

		if r.sourceVersion != "" && r.sourceVersion != obj.version {
			canUpdate = false
		}

		if r.sourceIdentify != "" && r.sourceIdentify != obj.identify {
			canUpdate = false
		}

		if canUpdate {
			obj.writeTime = time.Now().Unix()
			buf := SerialCacheObject(&obj)
			cacher.Put(key, buf)
		}
	}

	mge.notify(key,
		mergeData{true, false, nil, revalidStatus, mergeErrorErr},
		r.mergeToken)

	r.resp.Status = 200
	return
}

func (r *request) putChunkFragmentDone(fragmentSize int64, fragmentIndex int64) {
	cl := int64(0)
	i, err := strconv.ParseInt(r.chunkMergeDone, 10, 64)
	if err == nil && i > 0 {
		cl = i
	}

	headerSourceKey := base.GenObjectKey(comm.String2Bytes(r.headerSourceKey))

	if cl <= fragmentIndex*fragmentSize || cl > (fragmentIndex+1)*fragmentSize {
		logger.Error("put chunk fragment done failed, no expect content-length")
		cacher.Del(headerSourceKey)
	}

	// *******************************************************
	// TODO: 如果想要保证缓存的强一致性, 建议FusionCache提供一个事物接口.
	//	cacher.(*FusionCache).Transaction(func() {
	//		cacher.unsafeGet(...)
	//		parseCacheObject & update & SerialCacheObject
	//		cacher.unsafePut(...)
	//	})
	//

	headerValue, err := cacher.Get(headerSourceKey)
	if err != nil {
		return
	}

	obj := cacheObject{}
	err = ParseCacheObject(&obj, headerValue)
	if err != nil {
		cacher.Del(headerSourceKey)
		return
	}

	// cache object already changes
	if r.sourceVersion != "" && obj.version != r.sourceVersion {
		return
	}
	if r.sourceIdentify != "" && obj.identify != r.sourceIdentify {
		return
	}

	obj.contentLength = cl
	obj.writeTime = time.Now().Unix()
	buf := SerialCacheObject(&obj)
	cacher.Put(headerSourceKey, buf)
}

func (r *request) putChunkFragment() {
	defer r.done()

	if r.sourceKey == "" {
		r.resp.Headers[errorHeaderName] = noSourceKeyHeaderErr.Error()
		r.resp.Status = 400
		return
	}

	if r.clientToken == "" {
		r.resp.Headers[errorHeaderName] = noClientTokenHeaderErr.Error()
		r.resp.Status = 400
		return
	}

	r.sourceVersion = r.req.Headers[sourceVersionHeaderName]
	if r.sourceVersion == "" {
		r.resp.Headers[errorHeaderName] = noSourceVersionHeaderErr.Error()
		r.resp.Status = 400
		return
	}

	r.headerSourceKey = r.req.Headers[headerSourceKeyHeaderName]
	if r.headerSourceKey == "" {
		r.resp.Headers[errorHeaderName] = noHeaderSourceKeyHeaderErr.Error()
		r.resp.Status = 400
		return
	}

	r.chunkMergeToken = r.req.Headers[chunkMergeTokenHeaderName]
	if r.chunkMergeToken != "" {
		r.resp.Headers[errorHeaderName] = noChunkMergeTokenHeaderErr.Error()
		r.resp.Status = 400
		return
	}

	key := base.GenObjectKey(comm.String2Bytes(r.sourceKey))
	cmKey := r.headerSourceKey + "#" + r.sourceVersion

	// chunk传输中断, 删除chunkMergeNode, 并唤醒所有的读者
	r.chunkMergeError = r.req.Headers[chunkMergeErrorHeaderName]
	if r.chunkMergeError != "" {
		cacher.Del(key)
		cm.delete(cmKey, r.chunkMergeToken, errors.New(r.chunkMergeError))
		r.resp.Status = 200
		return
	}

	chunkFragmentSize := int64(0)
	r.chunkFragmentSize = r.req.Headers[chunkFragmentSizeHeaderName]
	if r.chunkFragmentSize != "" {
		i, err := strconv.ParseInt(r.chunkFragmentSize, 10, 64)
		if err == nil && i > 0 {
			chunkFragmentSize = i
		}
	}

	if chunkFragmentSize == 0 {
		r.resp.Headers[errorHeaderName] = invalidChunkFragmentSizeHeaderErr.Error()
		r.resp.Status = 400
		return
	}

	// headerFragment为0, 此处必定 > 0
	chunkFragmentIndex := int64(0)
	r.chunkFragmentIndex = r.req.Headers[chunkFragmentIndexHeaderName]
	if r.chunkFragmentIndex != "" {
		i, err := strconv.ParseInt(r.chunkFragmentIndex, 10, 64)
		if err == nil && i > 0 {
			chunkFragmentIndex = i
		}
	}

	if chunkFragmentIndex == 0 {
		r.resp.Headers[errorHeaderName] = invalidChunkFragmentIndexErr.Error()
		r.resp.Status = 400
		return
	}

	chunkMergeTimeoutMSec := config.ServerConf.ChunkMergeTimeoutMSec
	chunkMergeTimeoutMSecHeader := r.req.Headers[chunkMergeTimeoutMSecHeaderName]
	if chunkMergeTimeoutMSecHeader != "" {
		i, err := strconv.ParseInt(chunkMergeTimeoutMSecHeader, 10, 64)
		if err == nil && i > 0 {
			chunkMergeTimeoutMSec = i
		}
	}

	r.chunkMergeDone = r.req.Headers[chunkMergeDoneHeaderName]

	if r.req.Headers[cachableHeaderName] != "" {
		r.cachable = true
	}

	if r.req.Body == nil {
		r.req.Body = noBody
	}

	if r.cachable {
		cacher.Put(key, r.req.Body)
	}

	err := cm.update(cmKey, r.clientToken,
		chunkMergeData{
			fragmentSize:  chunkFragmentSize,
			fragmentIndex: chunkFragmentIndex,
			mergeDone:     r.chunkMergeDone,
		},
		chunkMergeTimeoutMSec)

	if err != nil {
		cacher.Del(key)
		cm.delete(cmKey, r.chunkMergeToken, err)
	}

	if r.chunkMergeDone != "" {
		r.putChunkFragmentDone(chunkFragmentSize, chunkFragmentIndex)
		cm.delete(cmKey, r.chunkMergeToken, nil)
	}

	r.resp.Status = 200
	return
}

func (r *request) getChunkFragment() {
	defer r.done()

	if r.sourceKey == "" {
		r.resp.Headers[errorHeaderName] = noSourceKeyHeaderErr.Error()
		r.resp.Status = 400
		return
	}

	if r.clientToken == "" {
		r.resp.Headers[errorHeaderName] = noClientTokenHeaderErr.Error()
		r.resp.Status = 400
		return
	}

	r.sourceVersion = r.req.Headers[sourceVersionHeaderName]
	if r.sourceVersion == "" {
		r.resp.Headers[errorHeaderName] = noSourceVersionHeaderErr.Error()
		r.resp.Status = 400
		return
	}

	r.headerSourceKey = r.req.Headers[headerSourceKeyHeaderName]
	if r.headerSourceKey == "" {
		r.resp.Headers[errorHeaderName] = noHeaderSourceKeyHeaderErr.Error()
		r.resp.Status = 400
		return
	}

	chunkFragmentSize := int64(0)
	r.chunkFragmentSize = r.req.Headers[chunkFragmentSizeHeaderName]
	if r.chunkFragmentSize != "" {
		i, err := strconv.ParseInt(r.chunkFragmentSize, 10, 64)
		if err == nil && i > 0 {
			chunkFragmentSize = i
		}
	}

	if chunkFragmentSize == 0 {
		r.resp.Headers[errorHeaderName] = invalidChunkFragmentSizeHeaderErr.Error()
		r.resp.Status = 400
		return
	}

	// headerFragment为0, 此处必定 > 0
	chunkFragmentIndex := int64(0)
	r.chunkFragmentIndex = r.req.Headers[chunkFragmentIndexHeaderName]
	if r.chunkFragmentIndex != "" {
		i, err := strconv.ParseInt(r.chunkFragmentIndex, 10, 64)
		if err == nil && i > 0 {
			chunkFragmentIndex = i
		}
	}

	if chunkFragmentIndex == 0 {
		r.resp.Headers[errorHeaderName] = invalidChunkFragmentIndexErr.Error()
		r.resp.Status = 400
		return
	}

	chunkMergeTimeoutMSec := config.ServerConf.ChunkMergeTimeoutMSec
	chunkMergeTimeoutMSecHeader := r.req.Headers[chunkMergeTimeoutMSecHeaderName]
	if chunkMergeTimeoutMSecHeader != "" {
		i, err := strconv.ParseInt(chunkMergeTimeoutMSecHeader, 10, 64)
		if err == nil && i > 0 {
			chunkMergeTimeoutMSec = i
		}
	}

	r.chunkMergeToken = r.req.Headers[chunkMergeTokenHeaderName]
	r.chunkMergeJoin = r.req.Headers[chunkMergeJoinHeaderName]
	r.chunkMergeRelease = r.req.Headers[chunkMergeReleaseHeaderName]

	if r.chunkMergeJoin != "" && r.chunkMergeRelease != "" {
		r.resp.Headers[errorHeaderName] = invalidChunkMergeOperationErr.Error()
		r.resp.Status = 400
		return
	}

	key := base.GenObjectKey(comm.String2Bytes(r.sourceKey))
	cmKey := r.headerSourceKey + "#" + r.sourceVersion

	if r.chunkMergeJoin != "" {
		chunkMergeToken, err := cm.join(cmKey, chunkMergeTimeoutMSec)
		if err != nil {
			headerKey := base.GenObjectKey(comm.String2Bytes(r.headerSourceKey))
			headerValue, err := cacher.Get(headerKey)
			if err != nil {
				r.resp.Headers[errorHeaderName] = "chunk client abort"
				r.resp.Status = 200
				return
			}

			totalLen := int64(-1)
			obj := cacheObject{}
			err = ParseCacheObject(&obj, headerValue)
			if err != nil {
				cacher.Del(headerKey)
				r.resp.Headers[errorHeaderName] = "found invalid cache object"
				r.resp.Status = 200
				return
			}

			totalLen = obj.contentLength
			if totalLen < 0 {
				cacher.Del(headerKey)
				r.resp.Headers[errorHeaderName] = "found invalid content-length in cache object"
				r.resp.Status = 200
				return
			}

			value, err := cacher.Get(key)
			if err != nil {
				r.resp.Headers[errorHeaderName] = "cache fragment lost"
				r.resp.Status = 200
				return
			}

			r.resp.Headers[cacheStatusHeaderName] = "HIT"
			r.resp.Headers[sourceContentLengthHeaderName] = strconv.FormatInt(totalLen, 10)
			r.resp.Body = value
			r.resp.Status = 200
			return
		}

		r.resp.Headers[chunkMergeTokenHeaderName] = chunkMergeToken
	}

	chunkMergeRelease := false
	defer func() {
		if chunkMergeRelease {
			cm.release(cmKey, r.chunkMergeToken)
		}
	}()

	if r.chunkMergeRelease != "" {
		chunkMergeRelease = true
	}

	// 没有加入chunkMerge, 简单的获取资源.
	if r.chunkMergeToken == "" {
		value, err := cacher.Get(key)
		if err != nil {
			r.resp.Status = 200
			return
		}

		r.resp.Headers[cacheStatusHeaderName] = "HIT"
		r.resp.Status = 200
		r.resp.Body = value
		return
	}

	c, cmData, err := cm.getCmData(cmKey, r.chunkMergeToken, chunkFragmentIndex, chunkMergeTimeoutMSec)
	if err != nil {
		headerKey := base.GenObjectKey(comm.String2Bytes(r.headerSourceKey))
		headerValue, err := cacher.Get(headerKey)
		if err != nil {
			r.resp.Headers[errorHeaderName] = "chunk client abort"
			r.resp.Status = 200
			return
		}

		totalLen := int64(-1)
		obj := cacheObject{}
		err = ParseCacheObject(&obj, headerValue)
		if err != nil {
			cacher.Del(headerKey)
			r.resp.Headers[errorHeaderName] = "found invalid cache object"
			r.resp.Status = 200
			return
		}

		totalLen = obj.contentLength
		if totalLen < 0 {
			cacher.Del(headerKey)
			r.resp.Headers[errorHeaderName] = "found invalid content-length in cache object"
			r.resp.Status = 200
			return
		}

		value, err := cacher.Get(key)
		if err != nil {
			r.resp.Headers[errorHeaderName] = "cache fragment lost"
			r.resp.Status = 200
			return
		}

		r.resp.Headers[cacheStatusHeaderName] = "HIT"
		r.resp.Headers[sourceContentLengthHeaderName] = strconv.FormatInt(totalLen, 10)
		r.resp.Body = value
		r.resp.Status = 200
		return
	}

	if c != nil {
		cmData = <-c
	}

	if cmData.err != nil {
		r.resp.Headers[errorHeaderName] = cmData.err.Error()
		r.resp.Status = 200
		return
	}

	if cmData.mergeDone != "" {
		r.resp.Headers[sourceContentLengthHeaderName] = cmData.mergeDone
		// 读取完最后一个Fragment, 直接离开chunkMerge
		if chunkFragmentIndex == cmData.fragmentIndex {
			chunkMergeRelease = true
		}
	}

	value, err := cacher.Get(key)
	if err != nil {
		r.resp.Status = 200
		return
	}

	r.resp.Headers[cacheStatusHeaderName] = "HIT"
	r.resp.Status = 200
	r.resp.Body = value
	return
}
